package org.raymond.iworks.study.mq.rabbitmq.stream;

import com.rabbitmq.client.*;
import com.rabbitmq.client.impl.DefaultExceptionHandler;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
/**
 * 流最初是为了覆盖现有队列无法满足的4种场景:
 * 1. large fan-out 大扇出
 *     当想要将相同的消息传递给多个订阅者时,用户目前必须为每个使用者绑定一个专用队列.如果消费者数量很大,
 * 这可能会降低效率,特别是在需要持久性和/或复制时.
 *     流将允许任意数量的消费者以非破坏性的方式使用来自同一队列的相同消息,从而不需要绑定多个队列.
 * 流消费者还可以从副本中读取数据,从而将读取负载分散到整个集群中.
 * 2. Replay(Time-travelling) 回放(穿越)
 *     由于当前所有RabbitMQ队列类型都具有破坏性的消费行为,即当消费者使用完消息后,消息将从队列中删除.
 * 因此不可能重新读取已被消费的消息.流将允许消费者在日志中的任何点附加并从那里读取.
 * 3, Throughput Performance 吞吐量性能
 *     没有任何持久队列类型能够提供与任何现有的基于日志的消息传递系统相竞争的吞吐量.流的设计以性能为主要目标.
 * 4, Large backlogs 大的积压
 *     大多数RabbitMQ队列都被设计为向空状态收敛,并进行了优化,当给定队列上有数百万条消息时,性能可能会更差.
 * 流的设计目的是以最小的内存开销的高效的方式存储大量数据.
 *
 * 单活动消费者:
 *     流的单活动消费者是RabbitMQ 3.11及后续版本中可用的特性.它在流上提供独占消费和消费连续性.
 * 当共享相同流和名称的几个消费者实例启用单个活动消费者时,这些实例中一次只有一个是活动的,因此将接收消息.
 * 其他实例将处于空闲状态.
 * 单个活跃消费者特性提供了两个好处:
 *     消息是按顺序处理的:一次只有一个消费者.
 *     保持消费连续性:如果活动的消费者停止或崩溃,这组其他消费者将接管.
 * 超级流:
 *     超级流是一种通过将大流划分为小流来扩展的方法.它们与单个活动消费者集成,以保持分区内的消息顺序.
 * 在RabbitMQ 3.11及更高版本中可以使用超级流.
 *     超级流是由独立的常规流组成的逻辑流.这是一种使用RabbitMQ Streams扩展发布和消费的方法:
 * 一个大的逻辑流被划分为多个分区流,将存储和流量分散到几个集群节点上.
 *     超级流仍然是一个逻辑实体:由于客户端库的智能,应用程序将其视为一个“大”流.超级流的拓扑结构基于AMQP 0.9.1模型,
 * 即它们之间的交换、队列和绑定.
 *     可以使用任何AMQP 0.9.1库或管理插件创建超级流的拓扑,它需要创建一个直接交换,“分区”流,并将它们绑定在一起.
 * 与单个流相比,超级流增加了复杂性,因此不应将其视为涉及流的所有用例的默认解决方案.只有当您确定达到了单个流的限制时,
 * 才考虑使用超级流.
 *
 * 流与经典队列的区别:
 * Feature                |Classic      |Stream
 * Non-durable queues     |yes          |no
 * Exclusivity            |yes          |no
 * Per message persistence|per message  |always
 * Membership changes	  |automatic    |manual
 * TTL	                  |yes          |no (but see Retention)
 * Queue length limits    |yes          |no (but see Retention)
 * Lazy behaviour         |yes          |inherent
 * Message priority       |yes          |no
 * Consumer priority      |yes          |no
 * Dead letter exchanges  |yes          |no
 * Adheres to policies	  |yes	        |(see Retention)
 * Reacts to memory alarms|yes          |no (uses minimal RAM)
 * Poison message handling|no	        |no
 * Global QoS Prefetch	  |yes	        |no
 *
 *     流不能被独占,不能用作临时队列
 *     流将所有数据直接存储在磁盘上,在消息被写入后,它不会使用任何内存,直到它被读取.可以说,流本质上是懒惰的
 *     流不支持全局预取(Global Qos)
 * Data Retention:
 *     流被实现为不可变的仅追加磁盘日志.这意味着日志将无限增长,直到磁盘耗尽.为了避免这种不希望出现的情况,
 * 可以设置每个流的保留配置,该配置将根据日志数据的总大小和/或年龄丢弃日志中最旧的数据.有两个参数控制流的保留.
 * max-age, max-length-bytes.
 * 留存率是基于每个片段进行评估的,所以还有一个参数会产生影响,那就是流的片段大小.流将始终保留至少一个段,
 * 只要该段包含至少一个消息.当使用代理提供的偏移量跟踪时,每个消费者的偏移量作为非消息数据保存在流本身中.
 * 性能特征:
 *     由于流在执行任何操作之前都会将所有数据保存到磁盘上,因此建议使用最快的磁盘.
 *     由于流的磁盘I/O繁重的特性,它们的吞吐量随着消息大小的增加而降低.
 *     就像仲裁队列一样,流也受到集群大小的影响.一个流的副本越多,它的吞吐量就越低,
 * 因为复制数据和达成共识需要做更多的工作
 */
public class StreamWorker {
    public static String EXCHANGE_NAME = "ex";
    public static String QUEUE_NAME = "stream";
    public static String ROUTING_KEY = "sk";

    public static void main(String[] args) throws Exception {
        Connection conn = connection();
        Channel channel = conn.createChannel();

        Map<String, Object> map = new HashMap<>();
        map.put("x-queue-type", "stream");
        // 固定参数,否则报错
        channel.queueDeclare(QUEUE_NAME, true, false, false, map);
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
        log.info("[*] Waiting for message");

        // 必须设置>0,否则报错
        channel.basicQos(1);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            log.info(" [x] Normal Received {}:{}", delivery.getEnvelope().getRoutingKey(), message);
            try{
                doWork(message);
            }catch (InterruptedException e){
                log.error("{}: {} will intterrupt.", e.getMessage(), Thread.currentThread().getName());
                Thread.currentThread().interrupt();
            }finally {
                log.info(" [x] Done");
                boolean multiple = false;
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), multiple);
            }
        };

        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
            log.info("ConsumerTag-{} cancel", consumerTag);
        });
    }
    private static void doWork(String task) throws InterruptedException {
        if(task.startsWith("interrupted")){
            Thread.currentThread().interrupt();
        }
        for (char ch: task.toCharArray()){
            if(ch=='.') Thread.sleep(10*1000);
        }
    }

    public static Connection connection() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.125");
        factory.setPort(5672);
        factory.setUsername("cloud");
        factory.setPassword("cloud123");
        factory.setVirtualHost("/cloud");

        factory.setExceptionHandler(new DefaultExceptionHandler(){
            @Override
            public void handleConsumerException(Channel channel, Throwable exception, Consumer consumer, String consumerTag, String methodName) {
                super.handleConsumerException(channel, exception, consumer, consumerTag, methodName);
                log.info("consumer exception: {}", exception.getMessage());
            }
        });
        return factory.newConnection();
    }
}
